/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.remoting.netty;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import com.alibaba.rocketmq.remoting.InvokeCallback;
import com.alibaba.rocketmq.remoting.common.SemaphoreReleaseOnlyOnce;
import com.alibaba.rocketmq.remoting.protocol.RemotingCommand;

/**
 * 异步请求应答封装
 * 
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2013-7-13
 */
public class ResponseFuture {
	private volatile RemotingCommand responseCommand;
	private volatile boolean sendRequestOK = true;
	private volatile Throwable cause;
	private final int opaque;
	private final long timeoutMillis;
	private final InvokeCallback invokeCallback;
	private final long beginTimestamp = System.currentTimeMillis();
	private final CountDownLatch countDownLatch = new CountDownLatch(1);

	// 保证信号量至多至少只被释放一次
	private final SemaphoreReleaseOnlyOnce once;

	// 保证回调的callback方法至多至少只被执行一次
	private final AtomicBoolean executeCallbackOnlyOnce = new AtomicBoolean(false);

	public ResponseFuture(int opaque, long timeoutMillis, InvokeCallback invokeCallback, SemaphoreReleaseOnlyOnce once) {
		this.opaque = opaque;
		this.timeoutMillis = timeoutMillis;
		this.invokeCallback = invokeCallback;
		this.once = once;
	}

	public void executeInvokeCallback() {
		if (invokeCallback != null) {
			// 先确保ResponseFuture.executeCallbackOnlyOnce的值为false并且成功更新为true
			// 由于executeCallbackOnlyOnce在初始化时为false，若更新失败说明该回调方法已经执行过了，故不在执行。
			if (this.executeCallbackOnlyOnce.compareAndSet(false, true)) {
				invokeCallback.operationComplete(this);
			}
		}
	}

	public void release() {
		if (this.once != null) {
			this.once.release();
		}
	}

	public boolean isTimeout() {
		long diff = System.currentTimeMillis() - this.beginTimestamp;
		return diff > this.timeoutMillis;
	}

	public RemotingCommand waitResponse(final long timeoutMillis) throws InterruptedException {
		this.countDownLatch.await(timeoutMillis, TimeUnit.MILLISECONDS);
		return this.responseCommand;
	}

	public void putResponse(final RemotingCommand responseCommand) {
		this.responseCommand = responseCommand;
		this.countDownLatch.countDown();
	}

	public long getBeginTimestamp() {
		return beginTimestamp;
	}

	public boolean isSendRequestOK() {
		return sendRequestOK;
	}

	public void setSendRequestOK(boolean sendRequestOK) {
		this.sendRequestOK = sendRequestOK;
	}

	public long getTimeoutMillis() {
		return timeoutMillis;
	}

	public InvokeCallback getInvokeCallback() {
		return invokeCallback;
	}

	public Throwable getCause() {
		return cause;
	}

	public void setCause(Throwable cause) {
		this.cause = cause;
	}

	public RemotingCommand getResponseCommand() {
		return responseCommand;
	}

	public void setResponseCommand(RemotingCommand responseCommand) {
		this.responseCommand = responseCommand;
	}

	public int getOpaque() {
		return opaque;
	}

	@Override
	public String toString() {
		return "ResponseFuture [responseCommand=" + responseCommand + ", sendRequestOK=" + sendRequestOK + ", cause=" + cause + ", opaque=" + opaque + ", timeoutMillis=" + timeoutMillis + ", invokeCallback=" + invokeCallback + ", beginTimestamp=" + beginTimestamp + ", countDownLatch=" + countDownLatch + "]";
	}
}
